package com.example.java.transformation;

import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

public class Main {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment
                .getExecutionEnvironment();
        ArrayList list = new ArrayList<Integer>();
        for (int i = 0; i<10;i++){
            list.add(i);
        }
        DataStreamSource<Integer> data = env.fromCollection(list);
        SplitStream<Integer> splitStream = data.split(new OutputSelector<Integer>() {
            @Override
            public Iterable<String> select(Integer value) {
                List<String> output = new ArrayList<String>();
                if (value % 2 == 0) {
                    output.add("even");
                } else {
                    output.add("odd");
                }
                return output;
            }
        });

//        splitStream.select("even").print();
        splitStream.select("odd").print();

        env.execute();

    }
}
